package com.platform.system.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

public class TencentMqttExample {

    public static void main(String[] args) throws MqttException, InterruptedException {
        // 从MQTT控制台获取接入点:
        // 通过Private Link实现VPC网络打通的用户, 使用内网接入点;
        // 通过公网访问的用户, 确保公网安全策略允许, 程序运行机器有公网接入;
        String serverUri = "tcp://mqtt-rkb747jx.iotcloud.tencentdevices.com:1883";
        // 合法的Client Identifier包含 数字0-9, 小写英文字母a-z, 以及大写英文字母A-Z, 总长度为1-23个字符
        // 参考 https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059
        String clientId = "NedgeGroup7777711";
        // 在控制台 --> 认证 Tab页创建账户, 复制用户名和密码
        String username = "xingyun";
        String password = "sk9b0b86e007466887";
        // 确认一级主题 "home" 在MQTT控制台已经创建
        String pubTopic = "device_sync";
        String[] topicFilters = new String[]{pubTopic, "home/#", "home/+"};
        int[] qos = new int[]{1, 1, 1};
        int total = 16;
        try (MqttClient client = new MqttClient(serverUri, clientId, new MemoryPersistence())) {
            client.setTimeToWait(3000);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
            options.setCleanSession(true);
            options.setAutomaticReconnect(true);
            MqttCallback callback = new MqttCallback(client, topicFilters, qos);
            client.setCallback(callback);
            client.connect(options);
            for (int i = 0; i < total; i++) {
                String msg = "Hello MQTT " + i;
                MqttMessage message = new MqttMessage(msg.getBytes(StandardCharsets.UTF_8));
                message.setQos(1);
                System.out.printf("Prepare to publish message %d%n", i);
                client.publish(pubTopic, message);
                System.out.printf("Published message %d%n", i);
                TimeUnit.SECONDS.sleep(3);
            }
            TimeUnit.SECONDS.sleep(3);
            client.disconnect();
        }
    }
    static class MqttCallback implements MqttCallbackExtended {
        private final MqttClient client;
        private final String[] topicFilters;
        private final int[] qos;
        public MqttCallback(MqttClient client, String[] topicFilters, int[] qos) {
            this.client = client;
            this.topicFilters = topicFilters;
            this.qos = qos;
        }
        @Override
        public void messageArrived(String topic, MqttMessage message) {
            System.out.printf("Message arrived, topic=%s, QoS=%d, Dup=%s, Retained=%s, content=[%s]%n",
                    topic, message.getQos(), message.isDuplicate(), message.isRetained(),
                    new String(message.getPayload(), StandardCharsets.UTF_8));
        }
        @Override
        public void connectionLost(Throwable cause) {
            System.out.println("connectionLost: " + cause.getMessage());
        }
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            System.out.println(reconnect ? "Reconnected" : "Connected" + " to " + serverURI);
            try {
                client.subscribe(topicFilters, qos);
                System.out.printf("Subscribed %d topics%n", topicFilters.length);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.printf("Delivery completed: packet-id=%d%n", token.getMessageId());
        }
    }
}
